package com.test.analyse;

import com.test.common.repository.perf.PerfReportBaseRepository;
import com.test.common.repository.perf.PerfReportRepository;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Properties;

@Slf4j
@Component
public class PerfReportAnalyse implements CommandLineRunner {

    @Resource
    private PerfReportRepository perfReportRepository;

    @Resource
    private PerfReportBaseRepository perfReportBaseRepository;

    @Resource(name = "analyse_config")
    private Config config;

    @Override
    public void run(String... args) throws Exception {

        log.info("启动解析程序");
        startAnalyse();
    }

    private void startAnalyse() {

        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, config.getKafkaGroup());
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBrokerUrl());
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, config.getConsumeThread());

        log.info("kafka stream配置：{}", props.toString());

        StreamsBuilder builder = new StreamsBuilder();

        KStream<String, String> textLines = builder.stream(config.getReportTopic());

        textLines.process(
                () -> new MyNewProcess(config.getAnalyseInterval(), perfReportRepository, perfReportBaseRepository));

        KafkaStreams streams = new KafkaStreams(builder.build(), props);

        streams.setUncaughtExceptionHandler((thread, throwable) -> {
            log.error("Kafka出现未处理的异常，被全局捕获到：{} - {}：", thread.getName(), throwable.getLocalizedMessage());
        });

        streams.setStateListener((newState, oldState) -> {
            log.info("Kafka状态变更：{} - {}", newState.name(), oldState.name());
        });

        streams.start();

        log.info("解析器启动");
    }
}
